package org.example.config;

import lombok.Data;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.example.constants.Constants;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * @author huangy
 * @description 消费者配置
 * @since 2022/11/10 10:36
 **/
@Data
@Configuration
public class RocketMQConsumerConfig implements CommandLineRunner {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Value("${message.queue.topic}")
    private String topic;

    @Resource
    private RocketUserConsumerListener userConsumerListener;

    @Resource
    private RocketRoomConsumerListener roomConsumerListener;

    @Resource
    private RocketCombatConsumerListener combatConsumerListener;

    public void messageListener() throws MQClientException {
        startConsumer(topic, Constants.RocketMqConstants.TAG_TYPE_USER, Constants.RocketMqConstants.GROUP_USER, userConsumerListener);
        startConsumer(topic, Constants.RocketMqConstants.TAG_TYPE_ROOM, Constants.RocketMqConstants.GROUP_ROOM, roomConsumerListener);
        startConsumer(topic, Constants.RocketMqConstants.TAG_TYPE_COMBAT, Constants.RocketMqConstants.GROUP_COMBAT, combatConsumerListener);
    }

    private void startConsumer(String topic, String tag, String group, MessageListenerConcurrently listener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //订阅topic
        consumer.subscribe(topic, tag);
        //注册监听器，收到消息时对消息进行处理
        consumer.registerMessageListener(listener);
        consumer.start();
    }

    @Override
    public void run(String... args) throws Exception {
        this.messageListener();
    }
}
